package com.meteor.raft.core;

import com.alibaba.fastjson.JSON;
import com.meteor.raft.config.PropertiesConfig;
import com.meteor.raft.persistent.RaftPeer;
import com.meteor.raft.util.NetUtils;
import com.ning.http.client.AsyncCompletionHandler;
import com.ning.http.client.Response;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.SortedBag;
import org.apache.commons.collections.bag.TreeBag;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.net.HttpURLConnection;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author gang.wang1903@xiaobao100.cn
 * @Date: 2019/7/15 8:08 PM
 * 节点处理类：存储所有节点信息：暂时不支持节点变更
 */
@Data
@Component
@Slf4j
public class RaftPeerManage {

    @Autowired
    private PropertiesConfig propertiesConfig;

    @Autowired
    private NetUtils netUtils;

    private AtomicLong localTerm = new AtomicLong(0L);

    private RaftPeer leader = null;

    private Map<String, RaftPeer> peers = new HashMap<String, RaftPeer>();

    private List<String> sites = new ArrayList<String>();

    @Value("${meteor.raft.is.standalone}")
    public  boolean isStandalone;

    public boolean ready = false;

    public RaftPeerManage() {

    }

    @PostConstruct
    public void init() {
        //初始化所有节点
        String ipString = propertiesConfig.getIp();
        List<String> ips = Arrays.asList(ipString.split(","));
        for(String ip : ips){
            RaftPeer raftPeer = new RaftPeer();
            raftPeer.setIp(ip);
            raftPeer.setState(RaftPeer.State.FOLLOWER);
            raftPeer.setVoteFor(null);
            peers.put(ip,raftPeer);
        }
    }


    /**
     * 获取leader
     * @return
     */
    public RaftPeer getLeader() {
        return leader;
    }

    /**
     * 删除节点
     * @param servers
     */
    public void remove(List<String> servers) {
        for (String server : servers) {
            peers.remove(server);
        }
    }


    /**
     * 获取自身节点信息
     * @return
     */
    public RaftPeer local() {
        RaftPeer peer = peers.get(netUtils.localServer());
        if (peer == null) {
            RaftPeer localPeer = new RaftPeer();
            localPeer.ip = netUtils.localServer();
            localPeer.term.set(localTerm.get());
            peers.put(localPeer.ip, localPeer);
            return localPeer;
        }
        if (peer == null) {
            throw new IllegalStateException("unable to find local peer: " + netUtils.localServer() + ", all peers: "
                    + Arrays.toString(peers.keySet().toArray()));
        }
        return peer;
    }


    /**
     * 重置所有节点投票
     */
    public void reset() {
        leader = null;
        for (RaftPeer peer : peers.values()) {
            peer.voteFor = null;
        }
    }

    /**
     * 获取初自己所有节点server
     * @return
     */
    public Set<String> allServersWithoutMySelf() {
        Set<String> servers = new HashSet<String>(peers.keySet());
        // exclude myself
        servers.remove(local().ip);
        return servers;
    }


    public Set<String> allServersIncludeMyself() {
        return peers.keySet();
    }
    /**
     * 处理主节点选举
     * @param candidate
     * @return
     */
    public RaftPeer decideLeader(RaftPeer candidate) {
        peers.put(candidate.ip, candidate);

        SortedBag ips = new TreeBag();
        int maxApproveCount = 0;
        String maxApprovePeer = null;
        for (RaftPeer peer : peers.values()) {
            if (StringUtils.isEmpty(peer.voteFor)) {
                continue;
            }

            ips.add(peer.voteFor);
            if (ips.getCount(peer.voteFor) > maxApproveCount) {
                maxApproveCount = ips.getCount(peer.voteFor);
                maxApprovePeer = peer.voteFor;
            }
        }

        if (maxApproveCount >= majorityCount()) {
            RaftPeer peer = peers.get(maxApprovePeer);
            peer.state = RaftPeer.State.LEADER;

            if (!Objects.equals(leader, peer)) {
                leader = peer;
               log.info("{} has become the LEADER", leader.ip);
            }
        }
        return leader;
    }

    public int majorityCount() {
        return peers.size() / 2 + 1;
    }

    public RaftPeer update(RaftPeer peer) {
        peers.put(peer.ip, peer);
        return peer;
    }


    public RaftPeer makeLeader(RaftPeer candidate) {
        if (!Objects.equals(leader, candidate)) {
            leader = candidate;
            //log.info("{} has become the LEADER, local: {}, leader: {}", leader.ip, JSON.toJSONString(local()), JSON.toJSONString(leader));
        }
        return update(candidate);
    }
}
